import org.apache.spark.sql.SparkSession
import java.util.Properties
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.log4j.{Level, Logger}

object WriteMySQL {
  def main(args: Array[String]): Unit = {
    // 设置日志级别为ERROR，以减少日志输出
    Logger.getLogger("org").setLevel(Level.ERROR)
    
    // 创建一个SparkSession对象，设置应用名称为"WriteMySQL"
    val spark = SparkSession.builder().appName("WriteMySQL").getOrCreate()
    
    // 设置两条数据的RDD
    val studentRDD = spark.sparkContext.parallelize(Array("Alice,Smith,22,Computer Science,2020-09-01")).map(_.split(","))
    
    // 设置模式信息
    val schema = StructType(List(
      StructField("first_name", StringType, true),
      StructField("last_name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("major", StringType, true),
      StructField("enrollment_date", StringType, true) // 注意这里使用StringType，因为我们将日期作为字符串处理
    ))

    // 创建Row对象的RDD
    val rowRDD = studentRDD.map { p =>
      // 将日期类型的字符串转换为LocalDate对象
      val enrollmentDate = p(4).trim
      Row(p(0).trim, p(1).trim, p(2).toInt, p(3).trim, enrollmentDate) // 直接将日期字符串作为字符串传递给Row
    }

    // 通过RDD和模式信息创建DataFrame
    val studentDF = spark.createDataFrame(rowRDD, schema)

    // 设置连接MySQL数据库的属性
    val prop = new Properties()
    prop.put("user", "root") // 设置用户名
    prop.put("password", "123456") // 设置密码
    prop.put("driver", "com.mysql.jdbc.Driver") // 设置MySQL驱动程序
    
    // 将DataFrame写入MySQL数据库中的"students"表中，采用追加模式
    studentDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparksql_practice", "students", prop)
    
    spark.stop() // 停止SparkSession
  }
}
